package david.java.flink_sql.conceptCommonApi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * @Description:
 * @See https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/common/
 * @Author: ZhaoDawei
 * @Date: Create in 10:16 上午 2022/4/7
 */
public class CreateTableFromFs {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // EnvironmentSettings settings = EnvironmentSettings
        //         .newInstance()
        //         .inStreamingMode()
        //         .build();
        // TableEnvironment tEnv = TableEnvironment.create(settings);

        Schema schema = Schema.newBuilder()
                //顺序要与csv一致
                .column("word", DataTypes.STRING())
                .column("count", DataTypes.INT())
                .build();

        tEnv.createTemporaryTable("source", TableDescriptor.forConnector("filesystem")
                .schema(schema)
                .option("path", "/Users/zhaodawei/code/david_flink/src/main/resources/data2")
                .format(FormatDescriptor.forFormat("csv")
                        .option("field-delimiter", ",")
                        .build())
                .build());

        tEnv.createTemporaryTable("sink", TableDescriptor.forConnector("filesystem")
                .schema(schema)
                .option("path", "/Users/zhaodawei/code/david_flink/src/main/resources/data1")
                .format(FormatDescriptor.forFormat("csv")
                        .build())
                .build());

        // The following code shows an example and the corresponding output for multiple-sinks plan using StatementSet.explain() method:
        StatementSet stmtSet = tEnv.createStatementSet();

        Table table1 = tEnv.from("source");
                // .where($("word").like("h%"));
        stmtSet.addInsert("sink", table1);

        String explain = stmtSet.explain();
        System.out.println(explain);


        stmtSet.execute().print();
        // tEnv.sqlQuery("select word, count from source").printSchema();
        // TableResult tableResult = tEnv.executeSql("select word, count from source");
        // tableResult.print();


    }
}
